internal/storage/badger.go (2)

428-453: Optimize cache update logic to avoid unnecessary allocations.

The cache update creates new big.Int instances when updating existing ranges, which is unnecessary.

 for chainStr, newRange := range chainRanges {
     existing, exists := bc.rangeCache[chainStr]
     if exists {
         // Update existing range
         if newRange.min.Cmp(existing.min) < 0 {
-            existing.min = newRange.min
+            existing.min = new(big.Int).Set(newRange.min)
         }
         if newRange.max.Cmp(existing.max) > 0 {
-            existing.max = newRange.max
+            existing.max = new(big.Int).Set(newRange.max)
         }
         existing.lastUpdated = time.Now()

---

669-680: Potential integer overflow when calculating new minimum block.

Adding 1 to blockNumber could theoretically overflow, though unlikely with big.Int.

Consider adding a comment explaining the logic:

 // The new minimum must be blockNumber + 1 or higher
+// Since we deleted all blocks <= blockNumber, the new minimum
+// is the next block if it exists, otherwise the cache entry is removed
 newMin := new(big.Int).Add(blockNumber, big.NewInt(1))

internal/source/s3.go (2)

625-681: Complex nested loop logic in streaming reader needs simplification.

The readRowGroupStreamingly function has deeply nested conditionals and could benefit from early returns.

Consider extracting the row processing logic:

func (s *S3Source) processParquetRow(row parquet.Row, blockMap map[uint64]bool, foundBlocks map[uint64]bool, results map[uint64]rpc.GetFullBlockResult) error {
    if len(row) < 8 {
        return nil // Skip invalid row
    }
    
    blockNum := row[1].Uint64()
    if !blockMap[blockNum] || foundBlocks[blockNum] {
        return nil // Skip unrequested or already found blocks
    }
    
    // Build and parse block data...
}

---

544-545: Remove outdated comments about block index.

The comments reference block index building which has been removed from the implementation.

-// Don't build block index immediately - build on demand to save memory
-// Block indices will be built lazily when needed
+// File is now ready for streaming reads

internal/storage/clickhouse.go (1)

1887-1910: Return semantics: ensure callers handle (nil, nil) range safely

MIN/MAX always return one row in ClickHouse; you’re correctly handling NULLs by returning (nil, nil). Verify all call sites guard against nil dereferences and treat “no staging data” distinctly from an error. Consider documenting this behavior above the method.

internal/orchestrator/validator.go (1)

23-29: Nil-guard worker dependency

NewValidator requires a non-nil worker. Add a defensive nil-check to fail fast in misconfigured tests/CLI wiring.

 func NewValidator(rpcClient rpc.IRPCClient, s storage.IStorage, w *worker.Worker) *Validator {
+    if w == nil {
+        panic("validator requires non-nil worker")
+    }
     return &Validator{
         rpc:     rpcClient,
         storage: s,
         worker:  w,
     }
 }

cmd/migrate_valid.go (2)

316-316: Reuse a single Worker instance

You construct two Workers (one for validator, one for migrator). Reuse one to share RPC concurrency limits and avoid duplicate semaphores.

-    validator := orchestrator.NewValidator(rpcClient, sourceConnector, worker.NewWorker(rpcClient))
+    w := worker.NewWorker(rpcClient)
+    validator := orchestrator.NewValidator(rpcClient, sourceConnector, w)
@@
     return &Migrator{
         batchSize:   batchSize,
         rpcClient:   rpcClient,
         source:      sourceConnector,
         destination: destinationConnector,
-        validator:   validator,
-        worker:      worker.NewWorker(rpcClient),
+        validator:   validator,
+        worker:      w,
     }

Also applies to: 323-331

---

235-236: Minor: structured context already present; drop Msgf

You already attach fields via Dur/Int. Prefer Msg with a static message to avoid mixing Msgf formatting with structured fields.

- log.Debug().Dur("duration", mapBuildDuration).Int("missing_blocks", len(missingBlocks)).Msgf("Worker %d: Identified missing blocks", workerID)
+ log.Debug().Dur("duration", mapBuildDuration).Int("missing_blocks", len(missingBlocks)).Int("worker_id", workerID).Msg("Identified missing blocks")

internal/storage/connector.go (1)

111-116: Typos in comments

• “orchestartor” → “orchestrator”
• “emphemeral” → “ephemeral”

internal/worker/worker.go (1)

425-435: Logging: incorrect last_block index

Use the last requested block, not results length, to index blockNumbers.

-      Str("first_block", blockNumbers[0].String()).
-      Str("last_block", blockNumbers[len(results)-1].String()).
+      Str("first_block", blockNumbers[0].String()).
+      Str("last_block", blockNumbers[len(blockNumbers)-1].String()).

internal/orchestrator/committer.go (3)

375-387: RPC latest block call: add timeout to avoid long stalls

Bound the RPC call with a short timeout to prevent blocking the commit loop.

-  latestBlock, err := c.rpc.GetLatestBlockNumber(ctx)
+  ctxLatest, cancel := context.WithTimeout(ctx, 3*time.Second)
+  defer cancel()
+  latestBlock, err := c.rpc.GetLatestBlockNumber(ctxLatest)
   if err != nil {
     return nil, fmt.Errorf("error getting latest block from RPC: %v", err)
   }

---

446-449: Metric should record the first missing block, not the first present block

Set MissedBlockNumbers to expectedBlockNumber (the first missing), not blocksData[0].

-      metrics.MissedBlockNumbers.Set(float64(blocksData[0].Block.Number.Int64()))
+      metrics.MissedBlockNumbers.Set(float64(expectedBlockNumber.Int64()))

---

520-529: Persist lastPublished in default mode after async publish

In default mode, you only update in-memory lastPublishedBlock. Consider persisting to OrchestratorStorage post-publish success for resilience across restarts.

I can provide a small helper to persist the highest block on success if desired.

internal/orchestrator/poller.go (3)

48-56: Add nil check for worker configuration

While there's a nil check for the worker config, it silently returns without any warning. This could lead to confusion if a nil worker is accidentally passed.

Add logging for better debugging:

 func WithPollerWorker(cfg *worker.Worker) PollerOption {
 	return func(p *Poller) {
 		if cfg == nil {
+			log.Warn().Msg("WithPollerWorker called with nil worker, ignoring")
 			return
 		}
 		
 		p.worker = cfg
 	}
 }

---

315-319: Consider using structured logging more consistently

The debug log uses structured fields which is good, but could be more consistent with field naming.

Use consistent field names:

 log.Debug().
-	Str("last_pending_block", lastPendingFetchBlock.String()).
-	Str("last_polled_block", lastPolledBlock.String()).
-	Str("last_requested_block", lastRequestedBlock.String()).
+	Str("last_pending_fetch_block", lastPendingFetchBlock.String()).
+	Str("last_polled_block", lastPolledBlock.String()).
+	Str("last_requested_block", lastRequestedBlock.String()).
 	Msgf("GetNextBlockRange for poller workers")

---

239-258: Consider memory optimization for large block ranges

The pollBlockData method could potentially handle large block ranges. The current implementation creates all block data in memory at once.

For very large block ranges, consider implementing streaming or chunked processing to reduce memory footprint. This could be particularly important if blocksPerPoll is set to a large value.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

internal/orchestrator/poller_test.go (1)

7-13: Replace placeholder with minimal, Worker-driven Poller tests (or gate with a build tag).

Placeholders erode coverage. Either:

• add a minimal table-driven test using the new Worker–Poller path, or
• gate the file with a build tag (e.g., //go:build integration) if the tests are temporarily integration-only.

I can scaffold a minimal Poller–Worker test harness against your new API if you share Poller/Worker constructors.

internal/orchestrator/committer_test.go (2)

30-53: Remove “Removed - test needs to be updated…” comment noise.

These comments add churn and can trigger format diffs. Delete them.

-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
-
-// Removed - test needs to be updated for new implementation
+// (intentionally left blank)

---

80-81: Don’t leave empty tests.

Skip with a reason or implement the behavior.

 func TestStartCommitter(t *testing.T) {
-}
+  t.Skip("TODO: implement start/stop lifecycle test after Worker–Poller wiring stabilizes")
+}

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

internal/orchestrator/poller.go (4)

145-151: Avoid busy-wait when there are no new blocks.

When ErrNoNewBlocks, the loop immediately continues and spins. Insert a small delay tied to triggerIntervalMs.

-          if err != nil {
-            if err != ErrNoNewBlocks {
-              log.Error().Err(err).Msg("Failed to get block range to poll")
-            }
-            continue
-          }
+          if err != nil {
+            if err != ErrNoNewBlocks {
+              log.Error().Err(err).Msg("Failed to get block range to poll")
+            } else {
+              // Avoid busy-waiting when chain tip hasn't advanced.
+              select {
+              case <-time.After(time.Duration(p.triggerIntervalMs) * time.Millisecond):
+              case <-pollCtx.Done():
+                return
+              }
+            }
+            continue
+          }

---

209-237: Fix wrong mutex and serialize Request to avoid races with concurrent calls.

lastRequestedBlock is updated while holding lastPolledBlockMutex, and concurrent Request() calls can trample each other or race with getNextBlockRange.

 func (p *Poller) Request(ctx context.Context, blockNumbers []*big.Int) []common.BlockData {
+  // Serialize request computations to avoid conflicting range math with pollers/other requests.
+  p.blockRangeMutex.Lock()
+  defer p.blockRangeMutex.Unlock()
   startBlock, endBlock := blockNumbers[0], blockNumbers[len(blockNumbers)-1]

   p.lastPolledBlockMutex.RLock()
   lastPolledBlock := new(big.Int).Set(p.lastPolledBlock)
   p.lastPolledBlockMutex.RUnlock()

   if startBlock.Cmp(lastPolledBlock) > 0 {
     log.Debug().Msgf("Requested block %s - %s is greater than last polled block %s, waiting for poller", startBlock.String(), endBlock.String(), lastPolledBlock.String())
     return nil
   }

   // If the requested end block exceeds, then truncate the block numbers list
   if endBlock.Cmp(lastPolledBlock) > 0 {
     lastPolledIndex := new(big.Int).Sub(lastPolledBlock, startBlock).Int64()
     blockNumbers = blockNumbers[:lastPolledIndex+1]
     log.Debug().Msgf("Truncated requested block range to %s - %s (last polled block: %s)", blockNumbers[0].String(), blockNumbers[len(blockNumbers)-1].String(), lastPolledBlock.String())
   }

   blockData, highestBlockNumber := p.pollBlockData(ctx, blockNumbers)
   if len(blockData) == 0 || highestBlockNumber == nil {
     return nil
   }

-  p.lastPolledBlockMutex.Lock()
-  p.lastRequestedBlock = new(big.Int).Set(highestBlockNumber)
-  p.lastPolledBlockMutex.Unlock()
+  p.lastRequestedBlockMutex.Lock()
+  p.lastRequestedBlock = new(big.Int).Set(highestBlockNumber)
+  p.lastRequestedBlockMutex.Unlock()
   return blockData
 }

#!/bin/bash
set -euo pipefail
# Inspect assignments to lastRequestedBlock to ensure they're protected by lastRequestedBlockMutex.
rg -n -C2 '\blastRequestedBlock\s*=' --type go

# Find any potentially unsafe uses of pollUntilBlock.Sign() that could deref nil.
rg -n -C2 'pollUntilBlock\.Sign\(' --type go

---

260-271: Skip errored RPC results in conversion.

Include result.Error guard to avoid staging empty/invalid data.

 func (p *Poller) convertPollResultsToBlockData(results []rpc.GetFullBlockResult) []common.BlockData {
   blockData := make([]common.BlockData, 0, len(results))
   for _, result := range results {
+    if result.Error != nil {
+      log.Error().Err(result.Error).Msgf("Error in block result for block %s", result.BlockNumber)
+      continue
+    }
     blockData = append(blockData, common.BlockData{
       Block:        result.Data.Block,
       Logs:         result.Data.Logs,
       Transactions: result.Data.Transactions,
       Traces:       result.Data.Traces,
     })
   }
   return blockData
 }

---

294-335: Don’t hold blockRangeMutex across RPC; also normalize “invalid range” as ErrNoNewBlocks.

Holding the lock while calling GetLatestBlockNumber can stall other workers and risks deadlocks if RPC hangs. Also, returning nil,nil for invalid range forces a downstream poll error; prefer ErrNoNewBlocks here.

 func (p *Poller) getNextBlockRange(ctx context.Context) ([]*big.Int, error) {
-  p.blockRangeMutex.Lock()
-  defer p.blockRangeMutex.Unlock()
-
   latestBlock, err := p.rpc.GetLatestBlockNumber(ctx)
   if err != nil {
     return nil, err
   }
 
+  p.blockRangeMutex.Lock()
+  defer p.blockRangeMutex.Unlock()
+
   p.lastPendingFetchBlockMutex.Lock()
   lastPendingFetchBlock := new(big.Int).Set(p.lastPendingFetchBlock)
   p.lastPendingFetchBlockMutex.Unlock()
@@
   endBlock := p.getEndBlockForRange(startBlock, latestBlock)
   if startBlock.Cmp(endBlock) > 0 {
     log.Debug().Msgf("Invalid range: start block %s is greater than end block %s, skipping", startBlock, endBlock)
-    return nil, nil
+    return nil, ErrNoNewBlocks
   }
@@
   p.lastPendingFetchBlockMutex.Lock()
   p.lastPendingFetchBlock = new(big.Int).Set(endBlock)
   p.lastPendingFetchBlockMutex.Unlock()
 
   return p.createBlockNumbersForRange(startBlock, endBlock), nil
 }

internal/orchestrator/poller.go (3)

69-91: Boundless init: consider setting pollUntilBlock to 0 to avoid nil checks.

This is optional given the fix below in reachedPollLimit, but it removes a whole class of nil pitfalls for “boundless” pollers.

Apply after constructing poller:

   poller := &Poller{
     chainId:           rpc.GetChainID(),
     rpc:               rpc,
     triggerIntervalMs: int64(triggerInterval),
     blocksPerPoll:     int64(blocksPerPoll),
     storage:           storage,
     parallelPollers:   config.Cfg.Poller.ParallelPollers,
   }
+  // For boundless polling, make "until" an explicit zero value instead of nil.
+  poller.pollUntilBlock = big.NewInt(0)

---

97-117: Initialization: surface storage readiness issues to callers.

Swallowing GetMaxBlockNumber errors can mask misconfigurations/downstream outages during startup. Either return an error from NewPoller (signature change) or record a health flag checked by Start() and exported via a readiness probe.

---

239-258: Minor logging nit.

Prefer structured logging to avoid fmt string churn and to ensure big.Ints are consistently stringified.

- log.Debug().Msgf("Polling %d blocks starting from %s to %s", len(blockNumbers), blockNumbers[0], blockNumbers[len(blockNumbers)-1])
+ log.Debug().
+   Int("block_count", len(blockNumbers)).
+   Str("start_block", blockNumbers[0].String()).
+   Str("end_block", blockNumbers[len(blockNumbers)-1].String()).
+   Msg("Polling blocks")

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

internal/orchestrator/reorg_handler_test.go (2)

7-8: Blocker: Don’t remove critical ReorgHandler tests—restore coverage now.

Eliminating these tests guts coverage for a high‑risk path. Re‑add tests in this PR before merge. Cover at least:

• Uses stored last reorg-checked block (>0) vs fallback to FromBlock
• Storage retrieval error → fallback to FromBlock
• Reorg detected vs no-reorg with new Worker/Poller wiring
• Boundary off‑by‑one at block edges

Run to confirm no remaining ReorgHandler/reorg tests elsewhere:

#!/bin/bash
set -euo pipefail
rg -nP --type=go -g '**/*_test.go' -C2 '(?i)\breorg(handler)?\b' || true
rg -nP --type=go '(?i)\b(last\s*reorg[-_ ]?checked|FromBlock|reorg\s*handler)\b' || true

---

10-12: Replace placeholder Skip with concrete subtest scaffolding.

Keep the file compiling but make next steps explicit and actionable.

-func TestReorgHandlerPlaceholder(t *testing.T) {
-	// Placeholder test to keep the test file valid
-	t.Skip("Reorg handler tests need to be rewritten for new implementation")
-}
+func TestReorgHandler_ReorgStartPoint(t *testing.T) {
+	t.Run("uses stored last reorg-checked block when >0", func(t *testing.T) {
+		t.Skip("TODO: inject mock store returning >0; assert start block used")
+	})
+	t.Run("falls back to FromBlock when store returns 0 or error", func(t *testing.T) {
+		t.Skip("TODO: mock store error/zero; assert fallback to config FromBlock")
+	})
+}
+
+func TestReorgHandler_ReorgDecisionFlow(t *testing.T) {
+	t.Run("no reorg detected -> no rollback", func(t *testing.T) {
+		t.Skip("TODO: wire Worker/Poller to stable chain; assert no rollback/re-validate")
+	})
+	t.Run("reorg detected -> triggers rollback/re-validate", func(t *testing.T) {
+		t.Skip("TODO: simulate fork; assert correct rollback and re-validate calls")
+	})
+	t.Run("boundary off-by-one at edges", func(t *testing.T) {
+		t.Skip("TODO: assert inclusive/exclusive bounds at start/end blocks")
+	})
+}

internal/worker/worker.go (4)

47-56: Good: rpcChunkSize guarded against zero

This addresses the SliceToChunks zero-size issue. Nice.

---

136-211: Failures mixed into successMap; wrong success detection and no error propagation

Errorful retries are inserted into successMap and later treated as successes; errors never reach failedResults, breaking fallback decisions (success becomes true with hidden failures).

Refactor to keep separate success and error maps, merge recursive results properly, and preserve original errors:

- // If only one block failed, retry once more
+ // If only one block failed, retry once more
   if len(failedBlocks) == 1 {
     retryResults := fetchFunc(ctx, failedBlocks)
     if len(retryResults) > 0 {
-      if retryResults[0].Error == nil {
-        successMap[failedBlocks[0].String()] = retryResults[0]
-      } else {
-        // Keep the error result
-        successMap[failedBlocks[0].String()] = rpc.GetFullBlockResult{
-          BlockNumber: failedBlocks[0],
-          Error:       retryResults[0].Error,
-        }
-      }
+      if retryResults[0].Error == nil {
+        successMap[failedBlocks[0].String()] = retryResults[0]
+      } else {
+        // Track failure; don't pollute successMap
+        errorMap := make(map[string]rpc.GetFullBlockResult)
+        errorMap[failedBlocks[0].String()] = retryResults[0]
+        // Build final results below will use errorMap
+        _ = errorMap
+      }
     }
   } else if len(failedBlocks) > 1 {
@@
-    var rwg sync.WaitGroup
-    var rwgMutex sync.Mutex
+    var rwg sync.WaitGroup
+    var rwgMutex sync.Mutex
+    errorMap := make(map[string]rpc.GetFullBlockResult)
@@
-      leftResults, _ := w.processChunkWithRetry(ctx, leftChunk, fetchFunc)
-      // Add results to map
-      for _, r := range leftResults {
+      leftSuccess, leftFailed := w.processChunkWithRetry(ctx, leftChunk, fetchFunc)
+      for _, r := range leftSuccess {
         if r.BlockNumber != nil {
           rwgMutex.Lock()
           successMap[r.BlockNumber.String()] = r
           rwgMutex.Unlock()
         }
       }
+      for _, r := range leftFailed {
+        if r.BlockNumber != nil {
+          rwgMutex.Lock()
+          errorMap[r.BlockNumber.String()] = r
+          rwgMutex.Unlock()
+        }
+      }
@@
-      rightResults, _ := w.processChunkWithRetry(ctx, rightChunk, fetchFunc)
-      // Add results to map
-      for _, r := range rightResults {
+      rightSuccess, rightFailed := w.processChunkWithRetry(ctx, rightChunk, fetchFunc)
+      for _, r := range rightSuccess {
         if r.BlockNumber != nil {
           rwgMutex.Lock()
           successMap[r.BlockNumber.String()] = r
           rwgMutex.Unlock()
         }
       }
+      for _, r := range rightFailed {
+        if r.BlockNumber != nil {
+          rwgMutex.Lock()
+          errorMap[r.BlockNumber.String()] = r
+          rwgMutex.Unlock()
+        }
+      }
@@
- // Build final results in original order
+ // Build final results in original order
   var finalResults []rpc.GetFullBlockResult
   var failedResults []rpc.GetFullBlockResult
   for _, block := range chunk {
     if result, ok := successMap[block.String()]; ok {
       finalResults = append(finalResults, result)
     } else {
-      // This should not happen as we have retried all failed blocks
-      failedResults = append(failedResults, rpc.GetFullBlockResult{
-        BlockNumber: block,
-        Error:       fmt.Errorf("failed to fetch block"),
-      })
+      // Prefer original failure if known
+      if em, ok := errorMap[block.String()]; ok {
+        failedResults = append(failedResults, em)
+      } else {
+        failedResults = append(failedResults, rpc.GetFullBlockResult{
+          BlockNumber: block,
+          Error:       fmt.Errorf("failed to fetch block"),
+        })
+      }
     }
   }

Note: define errorMap before single-retry branch and reuse it across paths.

---

405-409: Good: no sourceType shadowing on RPC fallback

Assignment (not :=) avoids the earlier logging mismatch.

---

214-273: Non-retry path drops error details and misclassifies failures

processChunk should also preserve error details and return them via failed, not embed them in success results.

- successMap := make(map[string]rpc.GetFullBlockResult)
+ successMap := make(map[string]rpc.GetFullBlockResult)
+ errorMap := make(map[string]rpc.GetFullBlockResult)
@@
- if result.Error == nil {
-   successMap[chunk[i].String()] = result
- }
+ if result.Error == nil {
+   successMap[chunk[i].String()] = result
+ } else {
+   errorMap[chunk[i].String()] = result
+ }
@@
- if result, ok := successMap[block.String()]; ok {
+ if result, ok := successMap[block.String()]; ok {
   finalResults = append(finalResults, result)
 } else {
-  // This should not happen as we have retried all failed blocks
-  failedResults = append(failedResults, rpc.GetFullBlockResult{
-    BlockNumber: block,
-    Error:       fmt.Errorf("failed to fetch block"),
-  })
+  if em, ok := errorMap[block.String()]; ok {
+    failedResults = append(failedResults, em)
+  } else {
+    failedResults = append(failedResults, rpc.GetFullBlockResult{
+      BlockNumber: block,
+      Error:       fmt.Errorf("failed to fetch block"),
+    })
+  }
 }

internal/orchestrator/poller.go (5)

290-292: Nil-deref in reachedPollLimit when pollUntilBlock is nil or zero

Guard nil and non-positive values.

-func (p *Poller) reachedPollLimit(blockNumber *big.Int) bool {
-  return blockNumber == nil || (p.pollUntilBlock.Sign() > 0 && blockNumber.Cmp(p.pollUntilBlock) >= 0)
-}
+func (p *Poller) reachedPollLimit(blockNumber *big.Int) bool {
+  if blockNumber == nil {
+    return true
+  }
+  if p.pollUntilBlock == nil || p.pollUntilBlock.Sign() == 0 {
+    return false
+  }
+  return blockNumber.Cmp(p.pollUntilBlock) >= 0
+}

---

300-304: Good: moved RPC call outside blockRangeMutex

Reduces contention and avoids lock-holding during network I/O.

---

145-151: Add backoff on ErrNoNewBlocks to avoid busy wait

The loop spins every 100ms without sleeping when there’s nothing to poll.

 if err != nil {
   if err != ErrNoNewBlocks {
     log.Error().Err(err).Msg("Failed to get block range to poll")
-  }
-  continue
+  } else {
+    select {
+    case <-time.After(100 * time.Millisecond):
+    case <-pollCtx.Done():
+      return
+    }
+  }
+  continue
 }

---

209-237: Synchronize Request with range planning and update correct state

Concurrent Request calls can race with getNextBlockRange. Serialize with blockRangeMutex.

 func (p *Poller) Request(ctx context.Context, blockNumbers []*big.Int) []common.BlockData {
+  p.blockRangeMutex.Lock()
+  defer p.blockRangeMutex.Unlock()
   startBlock, endBlock := blockNumbers[0], blockNumbers[len(blockNumbers)-1]
   ...
   p.lastRequestedBlockMutex.Lock()
   p.lastRequestedBlock = new(big.Int).Set(highestBlockNumber)
   p.lastRequestedBlockMutex.Unlock()
   return blockData
 }

---

260-271: Skip errored RPC results when converting to BlockData

Avoid staging invalid data; log and continue.

 for _, result := range results {
+  if result.Error != nil {
+    log.Error().Err(result.Error).Msgf("Error in block result for block %s", result.BlockNumber)
+    continue
+  }
   blockData = append(blockData, common.BlockData{
     Block:        result.Data.Block,
     Logs:         result.Data.Logs,
     Transactions: result.Data.Transactions,
     Traces:       result.Data.Traces,
   })
 }